package com.niit.sparkAnalyze.sparkStreaming
//各个类别订单的总数 需求（4
object orderCounts {
  def orderCounts(rdd: org.apache.spark.rdd.RDD[(String, String, Int, String, String)]): String = {
    if (!rdd.isEmpty()) {
      val orderCounts = rdd.map(record => {
        val orderCategory = record._1
        (orderCategory, 1)
      }).reduceByKey(_ + _) // 每个订单类别的数量累加

      val orderNumValueMap = rdd.map(record => record._1 -> record._3.toLong).collectAsMap()
      val res_str = orderCounts.map { case (category, count) =>
        val orderNumValue = orderNumValueMap.getOrElse(category, 0L)
        s"""[$category:${count * orderNumValue}]"""
      }.collect().mkString(",")


      if(!rdd.isEmpty())
      println("(sparkStreaming)" + res_str)
      res_str
    } else {
      "RDD is empty"
    }
  }
}
